﻿///////////////////////////////////////////////////////////////////////////////
//
//  This source is part of the  "Managed Task Framework.NET Runtime"  (MTF.NET Runtime).
//
//
//  Copyright (c) 2011, Multicore Consulting Limited (UK)
//  All rights reserved.
//  http://www.multicoreconsulting.co.uk
//
//  Redistribution and use in source and binary forms, with or without
//  modification, are permitted provided that the following conditions
//  are met:
//
//  -  Redistributions of source code must retain the above copyright
//     notice, this list of conditions and the following disclaimer.
//  -  Redistributions in binary form must reproduce the above copyright
//     notice, this list of conditions and the following disclaimer in
//     the documentation and/or other materials provided with the
//     distribution.
//  -  Neither the name of Multicore Consulting Limited nor the names of
//     its contributors may be used to endorse or promote products
//     derived from this software without specific prior written
//     permission.
//
//  THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
//  "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
//  LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS
//  FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE
//  COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT,
//  INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING,
//  BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
//  LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
//  CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
//  LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
//  ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
//  POSSIBILITY OF SUCH DAMAGE.
//
///////////////////////////////////////////////////////////////////////////////
//#define QUEUE_TRACE
//#define SEQUENCE_TRACE

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using MTFOrg.MTF.MTFCore.Messages;
using System.Diagnostics;

namespace MTFOrg.MTF.MTFCore
{
    /// <summary>
    /// Available queue types that this queue object can provide events from.
    /// </summary>
    [MTFRuntime.IsPropertyEnumeration()]
    public enum QueueType
    {
        NewItemQueue,		// Event contains a new item
        WrittenItemQueue	// Event contains a previously written item
    }

    /// <summary>
    /// Class implementing the queue object.
    /// </summary>
    partial class Queue
    {
        protected class WrittenItem
        {
            public object Item;
            public int ThreadAffinity;

            public WrittenItem(object item, int threadAffinity)
            {
                Item = item;
                ThreadAffinity = threadAffinity;
            }
        }

        protected class QueuedQuery
        {
            private int m_LinkNum;
            private int m_ProducerSlots;

            public int LinkNum
            {
                get { return m_LinkNum; }
                set { m_LinkNum = value; }
            }

            public int ProducerSlots
            {
                get { return m_ProducerSlots; }
                set { m_ProducerSlots = value; }
            }

            public QueuedQuery(int linkNum, int producerSlots)
            {
                m_LinkNum = linkNum;
                m_ProducerSlots = producerSlots;
            }
        }

        protected class QueueState
        {
            /// <summary>
            /// Queue of available new items.
            /// </summary>
            protected Queue<object> m_NewItems;

            /// <summary>
            /// Queue of available written items.
            /// </summary>
            protected SequencedQueue<WrittenItem> m_WrittenItems;

            /// <summary>
            /// Queue of link numbers waiting for a new item.
            /// </summary>
            protected Queue<QueuedQuery> m_QueuedForNewItem;

            /// <summary>
            /// Queue of link numbers waiting for a written item.
            /// </summary>
            protected Queue<QueuedQuery> m_QueuedForWrittenItem;

            /// <summary>
            /// Current sequence number.
            /// </summary>
            protected int m_SequenceNumber;

            /// <summary>
            /// Type of the object held in the queues.
            /// </summary>
            protected Type m_ObjectType;

            /// <summary>
            /// Gets the new items queue.
            /// </summary>
            public Queue<object> NewItems
            {
                get { return m_NewItems; }
            }

            /// <summary>
            /// Gets the written items queue.
            /// </summary>
            public SequencedQueue<WrittenItem> WrittenItems
            {
                get { return m_WrittenItems; }
                set { m_WrittenItems = value; }
            }

            /// <summary>
            /// Gets the queued for new item queue.
            /// </summary>
            public Queue<QueuedQuery> QueuedForNewItem
            {
                get { return m_QueuedForNewItem; }
            }

            /// <summary>
            /// Gets the queued for written item queue.
            /// </summary>
            public Queue<QueuedQuery> QueuedForWrittenItem
            {
                get { return m_QueuedForWrittenItem; }
            }

            /// <summary>
            /// Gets or sets the current sequence number.
            /// </summary>
            public int SequenceNumber
            {
                get { return m_SequenceNumber; }
                set { m_SequenceNumber = value; }
            }

            /// <summary>
            /// Constructor
            /// </summary>
            /// <param name="numItems">Number of items held by the queue</param>
            /// <param name="objectType">Type of item held in the queue</param>
            public QueueState(int numItems, Type objectType, bool isOrdered)
            {
                // Create the queues
                m_QueuedForNewItem = new Queue<QueuedQuery>();
                m_QueuedForWrittenItem = new Queue<QueuedQuery>();
                m_NewItems = new Queue<object>();
                m_WrittenItems = new SequencedQueue<WrittenItem>(numItems, isOrdered);

                m_ObjectType = objectType;

                // Initialise the new items queue with all the available items
                for (int n = 0; n < numItems; n++)
                    m_NewItems.Enqueue(null);
            }
        }

        /// <summary>
        /// Number of objects this queue holds.
        /// </summary>
        int m_NumObjects;

        /// <summary>
        /// Type of objects that this queue holds.
        /// </summary>
        Type m_ObjectType;

        /// <summary>
        /// Whether this queue enables thread-affinity.
        /// </summary>
        bool m_HasAffinity;

        /// <summary>
        /// Whether reads from the queue should be ordered sequentially.
        /// </summary>
        bool m_IsOrdered;

        /// <summary>
        /// Invoked when the task is initialized.
        /// </summary>
        public override void OnInitialize()
        {
            base.OnInitialize();

            // Get property information
            m_NumObjects = (int)GetProperty("NumObjects");
            m_ObjectType = (Type)GetProperty("ObjectType");
            m_HasAffinity = (bool)GetProperty("HasAffinity");
            object oIsOrdered = GetProperty("IsOrdered");
            m_IsOrdered = oIsOrdered != null ? (bool)oIsOrdered : false;

            // Hook up to the producer-consumer event notifications
            m_ProducerTask.QueryReceived += new ProducerConsumer.OnQuery(OnQueryReceived);
            m_ProducerTask.ReleaseReceived += new ProducerConsumer.OnRelease(OnReleaseReceived);
        }

        /// <summary>
        /// Gets the tag name for the specified connection.
        /// </summary>
        /// <param name="producerConnection">Producer connection to get tag for</param>
        /// <returns>Tag for given producer connection</returns>
        string GetTagName(MTFRuntime.IConnection producerConnection)
        {
            MTFRuntime.ConnectionProperty tagNameProp = producerConnection.GetProperty("TagName") as MTFRuntime.ConnectionProperty;
            string tagName;
            if (tagNameProp != null)
            {
                tagName = (string)tagNameProp.Value;
            }
            else
            {
                tagName = Name;
            }

            return tagName;
        }

        /// <summary>
        /// Handles received query.
        /// </summary>
        /// <param name="producerLinkNum">Link that query was received on</param>
        /// <param name="consumerSlots">Consumer slot states (not used)</param>
        /// <param name="producerSlots">Slot states</param>
        /// <param name="elementState">State associated with the current element</param>
        /// <param name="elementNumber">Number of the current element</param>
        public void OnQueryReceived(int producerLinkNum, Slots slots, int consumerSlot, int producerSlot, ref object elementState, int elementNumber)
        {
            QueueState queueState = elementState as QueueState;
            if (queueState == null)
            {
                elementState = queueState = new QueueState(m_NumObjects, m_ObjectType, m_IsOrdered);
            }

            // Get properties for connection
            MTFRuntime.IConnection producerConnection = m_ProducerTask.GetProducerConnectionFromLinkNumber(producerLinkNum);
            MTFRuntime.ConnectionProperty queueTypeProp = producerConnection.GetProperty("QueueType");
            QueueType queueType;
            if (queueTypeProp != null)
            {
                queueType = (QueueType)producerConnection.GetProperty("QueueType").Value;
            }
            else
            {
                throw new Exception("QueueType property was not specified on connection to " + FullPath);
            }
            string tagName = GetTagName(producerConnection);

            // TODO: Look in appropriate queue for a buffer and produce an event if the buffer is available and
            // move the buffer to a busy queue, otherwise queue this query until a buffer becomes available.
            if (queueType == QueueType.NewItemQueue)
            {
                if (queueState.NewItems.Count > 0)
                {
                    // Get new item
                    object newItem = queueState.NewItems.Dequeue();
#if SEQUENCE_TRACE
                    Debug.Print(Name + "SN: " + queueState.SequenceNumber.ToString());
#endif
                    m_ProducerTask.SendEvent(producerLinkNum, producerSlot, new EventDictionary(tagName, new QueueEvent(newItem, QueueType.NewItemQueue, m_HasAffinity, queueState.SequenceNumber++)), elementNumber);
#if SENDEVENT_TRACE
                    Debug.Print("{0}[{1}] link {2}", Name, elementNumber, producerLinkNum);
#endif
#if QUEUE_TRACE
					Debug.Print(Name + ": N(Q) -N  " + string.Format("(N={0} W={1} QN={2} QW={3})", queueState.NewItems.Count, queueState.WrittenItems.Count, queueState.QueuedForNewItem.Count, queueState.QueuedForWrittenItem.Count));
#endif
                }
                else
                {
                    // Add request to queue
                    queueState.QueuedForNewItem.Enqueue(new QueuedQuery(producerLinkNum, producerSlot));
#if QUEUE_TRACE
					Debug.Print(Name + ": N(Q) +QN " + string.Format("(N={0} W={1} QN={2} QW={3})", queueState.NewItems.Count, queueState.WrittenItems.Count, queueState.QueuedForNewItem.Count, queueState.QueuedForWrittenItem.Count));
#endif
                }
            }
            else if (queueType == QueueType.WrittenItemQueue)
            {
                // Add request to queue
                queueState.QueuedForWrittenItem.Enqueue(new QueuedQuery(producerLinkNum, producerSlot));
#if QUEUE_TRACE
				Debug.Print(Name + ": W(Q) +QW " + string.Format("(N={0} W={1} QN={2} QW={3})", queueState.NewItems.Count, queueState.WrittenItems.Count, queueState.QueuedForNewItem.Count, queueState.QueuedForWrittenItem.Count));
#endif
                while (queueState.QueuedForWrittenItem.Count > 0 && queueState.WrittenItems.Count > 0)
                {
                    // Get the queued query
                    QueuedQuery queuedQuery = queueState.QueuedForWrittenItem.Dequeue();
                    MTFRuntime.IConnection queuedProducerConnection = m_ProducerTask.GetProducerConnectionFromLinkNumber(queuedQuery.LinkNum);
                    string queuedTagName = GetTagName(queuedProducerConnection);

                    // Get written item
                    WrittenItem writtenItem = queueState.WrittenItems.Dequeue();
                    QueueEvent newEvent = new QueueEvent(writtenItem.Item, QueueType.WrittenItemQueue, m_HasAffinity, 0);
                    newEvent.ThreadAffinity = writtenItem.ThreadAffinity;

                    // Send the written item to the query originator
                    m_ProducerTask.SendEvent(producerLinkNum, producerSlot, new EventDictionary(tagName, newEvent), elementNumber);
#if SENDEVENT_TRACE
                    Debug.Print("{0}[{1}] link {2}", Name, elementNumber, producerLinkNum);
#endif
#if QUEUE_TRACE
				    Debug.Print(Name + ": W(Q) -W  " + string.Format("(N={0} W={1} QN={2} QW={3})", queueState.NewItems.Count, queueState.WrittenItems.Count, queueState.QueuedForNewItem.Count, queueState.QueuedForWrittenItem.Count));
#endif
                }
            }
#if DEBUG
            else
            {
                throw new NotImplementedException("Queue received release of event from an unknown queue type.");
            }
#endif
        }

        /// <summary>
        /// Handles received release.
        /// </summary>
        /// <param name="producerLinkNum">Link that query was received on</param>
        /// <param name="consumerSlots">Consumer slot states (not used)</param>
        /// <param name="producerSlots">Slot states</param>
        /// <param name="eventToRelease">The event to be released</param>
        /// <param name="isRollback">Whether the release was a roll-back</param>
        /// <param name="elementState">State associated with the current element</param>
        /// <param name="elementNumber">Number of the current element</param>
        /// <returns>Whether to complete the current slot set</returns>
        public bool OnReleaseReceived(int producerLinkNum, Slots slots, int consumerSlot, int producerSlot, Event eventToRelease, bool isRollback, bool deletedSlot, ref object elementState, int elementNumber)
        {
#if DEBUG
            if (!(eventToRelease is EventDictionary))
                throw new Exception("Queue is attempting to release an unsupported event type");
#endif

            MTFRuntime.IConnection producerConnection = m_ProducerTask.GetProducerConnectionFromLinkNumber(producerLinkNum);
            string tagName = GetTagName(producerConnection);
            QueueState queueState = elementState as QueueState;
            QueueEvent queueEvent = (eventToRelease as EventDictionary).GetEvent(tagName) as QueueEvent;

#if DEBUG
            if (queueEvent == null)
                throw new Exception("Queue is attempting to release a non-queue event.");
#endif

            if (!queueEvent.IsCanceled())
            {
                // Put the associated buffer back into circulation in the opposite queue or if there are
                // queued queries then produce an event with this item immediately.
                if (queueEvent.SourceQueue == QueueType.NewItemQueue)
                {
                    // This came from the new item queue and is therefore returned as a written item.
                    queueState.WrittenItems.Enqueue(new WrittenItem(queueEvent.Item, queueEvent.ThreadAffinity), queueEvent.SequenceNumber);
#if QUEUE_TRACE
				    Debug.Print(Name + ": N(R) +W  " + string.Format("(N={0} W={1} QN={2} QW={3})", queueState.NewItems.Count, queueState.WrittenItems.Count, queueState.QueuedForNewItem.Count, queueState.QueuedForWrittenItem.Count));
#endif

                    // While there are written items and corresponding queries
                    while (queueState.QueuedForWrittenItem.Count > 0 && queueState.WrittenItems.Count > 0)
                    {
                        // Get the queued query
                        QueuedQuery queuedQuery = queueState.QueuedForWrittenItem.Dequeue();
                        MTFRuntime.IConnection queuedProducerConnection = m_ProducerTask.GetProducerConnectionFromLinkNumber(queuedQuery.LinkNum);
                        string queuedTagName = GetTagName(queuedProducerConnection);

                        // Get the written item
                        WrittenItem nextWrittenItem = queueState.WrittenItems.Dequeue();
                        QueueEvent writtenEvent = new QueueEvent(nextWrittenItem.Item, QueueType.WrittenItemQueue, m_HasAffinity, 0);
                        writtenEvent.ThreadAffinity = nextWrittenItem.ThreadAffinity;

                        // Send the written item to the query originator
                        m_ProducerTask.SendEvent(queuedQuery.LinkNum, queuedQuery.ProducerSlots, new EventDictionary(queuedTagName, writtenEvent), elementNumber);
#if SENDEVENT_TRACE
                        Debug.Print("{0}[{1}] link {2}", Name, elementNumber, queuedQuery.LinkNum);
#endif
#if QUEUE_TRACE
					    Debug.Print(Name + ": N(R) -QW " + string.Format("(N={0} W={1} QN={2} QW={3})", queueState.NewItems.Count, queueState.WrittenItems.Count, queueState.QueuedForNewItem.Count, queueState.QueuedForWrittenItem.Count));
#endif
                    }
                }
                else if (queueEvent.SourceQueue == QueueType.WrittenItemQueue)
                {
                    // This came from the written item queue so can be destroyed and a new item created.
                    if (queueState.QueuedForNewItem.Count > 0)
                    {
                        QueuedQuery queuedQuery = queueState.QueuedForNewItem.Dequeue();
                        MTFRuntime.IConnection queuedProducerConnection = m_ProducerTask.GetProducerConnectionFromLinkNumber(queuedQuery.LinkNum);
                        string queuedTagName = GetTagName(queuedProducerConnection);
#if SEQUENCE_TRACE
                        Debug.Print(Name + "SN: " + queueState.SequenceNumber.ToString());
#endif
                        m_ProducerTask.SendEvent(queuedQuery.LinkNum, queuedQuery.ProducerSlots, new EventDictionary(queuedTagName, new QueueEvent(null, QueueType.NewItemQueue, m_HasAffinity, queueState.SequenceNumber++)), elementNumber);
#if SENDEVENT_TRACE
                        Debug.Print("{0}[{1}] link {2}", Name, elementNumber, queuedQuery.LinkNum);
#endif
#if QUEUE_TRACE
						Debug.Print(Name + ": W(R) -QN " + string.Format("(N={0} W={1} QN={2} QW={3})", queueState.NewItems.Count, queueState.WrittenItems.Count, queueState.QueuedForNewItem.Count, queueState.QueuedForWrittenItem.Count));
#endif
                    }
                    else
                    {
                        queueState.NewItems.Enqueue(null);
#if QUEUE_TRACE
						Debug.Print(Name + ": W(R) +N  " + string.Format("(N={0} W={1} QN={2} QW={3})", queueState.NewItems.Count, queueState.WrittenItems.Count, queueState.QueuedForNewItem.Count, queueState.QueuedForWrittenItem.Count));
#endif
                    }
                }
#if DEBUG
                else
                {
                    throw new NotImplementedException("Queue received release of event from an unknown queue type.");
                }
#endif
            }
            else
            {
                // Put the associated buffer back into circulation in the same queue or if there are
                // queued queries then produce an event with this item immediately.
                if (queueEvent.SourceQueue == QueueType.WrittenItemQueue)
                {
                    // This came from the written item queue and is therefore returned as a written item.
                    if (queueEvent.IsCanceledToFront())
                        queueState.WrittenItems.Prequeue(new WrittenItem(queueEvent.Item, queueEvent.ThreadAffinity));
                    else
                    {
#if SEQUENCE_TRACE
                        Debug.Print(Name + "SN: " + queueState.SequenceNumber.ToString());
#endif
                        queueState.WrittenItems.Enqueue(new WrittenItem(queueEvent.Item, queueEvent.ThreadAffinity), queueState.SequenceNumber++);
                    }
#if QUEUE_TRACE
					Debug.Print(Name + ": N(X) +W  " + string.Format("(N={0} W={1} QN={2} QW={3})", queueState.NewItems.Count, queueState.WrittenItems.Count, queueState.QueuedForNewItem.Count, queueState.QueuedForWrittenItem.Count));
#endif

                    // While there are written items and corresponding queries
                    while (queueState.QueuedForWrittenItem.Count > 0 && queueState.WrittenItems.Count > 0)
                    {
                        // Get the queued query
                        QueuedQuery queuedQuery = queueState.QueuedForWrittenItem.Dequeue();
                        MTFRuntime.IConnection queuedProducerConnection = m_ProducerTask.GetProducerConnectionFromLinkNumber(queuedQuery.LinkNum);
                        string queuedTagName = GetTagName(queuedProducerConnection);

                        // Get the written item
                        WrittenItem nextWrittenItem = queueState.WrittenItems.Dequeue();
                        QueueEvent writtenEvent = new QueueEvent(nextWrittenItem.Item, QueueType.WrittenItemQueue, m_HasAffinity, 0);
                        writtenEvent.ThreadAffinity = nextWrittenItem.ThreadAffinity;

                        // Send the written item to the query originator
                        m_ProducerTask.SendEvent(queuedQuery.LinkNum, queuedQuery.ProducerSlots, new EventDictionary(queuedTagName, writtenEvent), elementNumber);
#if SENDEVENT_TRACE
                        Debug.Print("{0}[{1}] link {2}", Name, elementNumber, queuedQuery.LinkNum);
#endif
#if QUEUE_TRACE
						Debug.Print(Name + ": N(X) -QW " + string.Format("(N={0} W={1} QN={2} QW={3})", queueState.NewItems.Count, queueState.WrittenItems.Count, queueState.QueuedForNewItem.Count, queueState.QueuedForWrittenItem.Count));
#endif
                    }
                }
                else if (queueEvent.SourceQueue == QueueType.NewItemQueue)
                {
                    // This came from the new item queue so can be destroyed and a new item created.
                    queueState.WrittenItems.Skip(queueEvent.SequenceNumber);
                    if (queueState.QueuedForNewItem.Count > 0)
                    {
                        QueuedQuery queuedQuery = queueState.QueuedForNewItem.Dequeue();
                        MTFRuntime.IConnection queuedProducerConnection = m_ProducerTask.GetProducerConnectionFromLinkNumber(queuedQuery.LinkNum);
                        string queuedTagName = GetTagName(queuedProducerConnection);
#if SEQUENCE_TRACE
                        Debug.Print(Name + "SN: " + queueState.SequenceNumber.ToString());
#endif
                        m_ProducerTask.SendEvent(queuedQuery.LinkNum, queuedQuery.ProducerSlots, new EventDictionary(queuedTagName, new QueueEvent(null, QueueType.NewItemQueue, m_HasAffinity, queueState.SequenceNumber++)), elementNumber);
#if SENDEVENT_TRACE
                        Debug.Print("{0}[{1}] link {2}", Name, elementNumber, queuedQuery.LinkNum);
#endif
#if QUEUE_TRACE
					    Debug.Print(Name + ": W(X) -QN " + string.Format("(N={0} W={1} QN={2} QW={3})", queueState.NewItems.Count, queueState.WrittenItems.Count, queueState.QueuedForNewItem.Count, queueState.QueuedForWrittenItem.Count));
#endif
                    }
                    else
                    {
                        queueState.NewItems.Enqueue(null);
#if QUEUE_TRACE
					    Debug.Print(Name + ": W(X) +N  " + string.Format("(N={0} W={1} QN={2} QW={3})", queueState.NewItems.Count, queueState.WrittenItems.Count, queueState.QueuedForNewItem.Count, queueState.QueuedForWrittenItem.Count));
#endif
                    }
                }
#if DEBUG
                else
                {
                    throw new NotImplementedException("Queue received release of event from an unknown queue type.");
                }
#endif
            }


            return true;
        }
    }
}